package jmind.core.rabbitmq;


import com.rabbitmq.client.*;
import jmind.base.lang.IProperties;
import jmind.base.util.GlobalConstants;
import org.perf4j.StopWatch;
import org.perf4j.slf4j.Slf4JStopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created by xieweibo on 2016/10/31.
 */
public abstract class RabbitmqConsumer implements com.rabbitmq.client.Consumer {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    private final Channel channel;
    private boolean autoAck ;

    public RabbitmqConsumer(String name, IProperties p, String queueName, String exchangeName, String routeKey, boolean autoAck) throws IOException, TimeoutException {
        logger.debug("name={},queueName={},exchangeName={},routeKey={}",name,queueName,exchangeName,routeKey);
        this.autoAck=autoAck;
        Connection connection = RabbitmqFactory.getFactory().getConnection(name, p);
        channel = connection.createChannel();
        // RabbitMQ客户端同时接受消息最大数量
        channel.basicQos(30);
       try{
           channel.queueDeclare(queueName, true, false, false, null);
       }catch (Exception e){
       }
       // channel.exchangeDeclare(exchangeName, type.name(),durable);

        // 绑定
        channel.queueBind(queueName, exchangeName, routeKey);
        channel.basicConsume(queueName, autoAck, this);
        if (connection != null) {
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    try {
                        channel.close();
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }


    }

    protected abstract boolean consume(Envelope envelope, byte[] body);

    @Override
    public void handleConsumeOk(String consumerTag) {

    }

    @Override
    public void handleCancelOk(String consumerTag) {

    }

    @Override
    public void handleCancel(String consumerTag) throws IOException {

    }

    @Override
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {

    }

    @Override
    public void handleRecoverOk(String consumerTag) {

    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        logger.debug("consume-{}={}",envelope.getRoutingKey(),new String(body, GlobalConstants.CHARSET_UTF8));
        StopWatch stopWatch = new Slf4JStopWatch("shell");
        boolean dealRet=false;
        try{
             dealRet=consume(envelope, body);
        }finally {
            stopWatch.stop(envelope.getRoutingKey());
            // ack
            if(!autoAck){ // 不自动ack
                if ( dealRet) {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }else{
                    // 消费失败，把此消息发送给下一个消费者
                    // requeue 为true发送给下一个消费者，false mq会把消息从队列移除
                    channel.basicReject(envelope.getDeliveryTag(),true);
                }
            }
        }



    }


    public Channel getChannel() {
        return channel;
    }
}
